package com.imooc.ad.mysql;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.imooc.ad.mysql.listener.AggregationListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * 与mysql连接，实现监听Binlog的启动、关闭
 */
@Slf4j
@Component
public class BinlogClient {

    private BinaryLogClient client;

    //注入配置信息、Binlog监听器
    private final BinlogConfig config;
    private final AggregationListener listener;

    @Autowired
    public BinlogClient(BinlogConfig config, AggregationListener listener) {
        this.config = config;
        this.listener = listener;
    }

    //连接数据库
    public void connect() {
        new Thread(() -> {
            //配置信息
            client = new BinaryLogClient(
                    config.getHost(), config.getPort(), config.getUsername(), config.getPassword()
            );
            //未指定监听文件，则默认从当前文件开始
            if (!StringUtils.isEmpty(config.getBinlogName()) && !config.getPosition().equals(-1L)) {
                client.setBinlogFilename(config.getBinlogName());
                client.setBinlogPosition(config.getPosition());
            }
            //注册监听器
            client.registerEventListener(listener);
            //连接mysql
            try {
                log.info("connecting to mysql start");
                client.connect();
                log.info("connecting to mysql done");
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }).start();
    }

    //关闭连接
    public void close() {
        try {
            client.disconnect();
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }
}
